/**
 * Copyright(c) 2013 Jade Techonologies Co., Ltd.
 *
 * History:
 *   14-8-4 上午11:29 Created by dwzhang
 */
package com.jade.framework.mom.rocketmq;

import java.io.Serializable;
import java.util.Iterator;
import java.util.Set;

import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.alibaba.rocketmq.client.producer.SendStatus;
import com.alibaba.rocketmq.common.message.Message;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.alibaba.rocketmq.remoting.exception.RemotingException;
import com.jade.framework.mom.MomConstants;
import com.jade.framework.mom.MomException;
import com.jade.framework.mom.MomMessage;
import com.jade.framework.mom.MomSendResult;
import org.apache.commons.lang3.SerializationUtils;
import org.apache.commons.lang3.StringUtils;

/**
 * RocketMQ的工具类。
 *
 * @author dwzhang
 * @version 1.0 14-8-4 上午11:29
 */
public class RocketmqUtils
{
    public static int toRmqDelayLevel(int level)
    {
        return level;
    }

    public static int toMomDelayLevel(int level)
    {
        return level;
    }

    public static MomException handleException(Throwable t)
    {
        if (t instanceof MQClientException) {
            return new MomException(MomConstants.ERR_CLIENT,
                    String.valueOf(((MQClientException) t).getResponseCode()) + ":" +
                            ((MQClientException) t).getErrorMessage(), t);
        }
        else if (t instanceof RemotingException) {
            return new MomException(MomConstants.ERR_REMOTING, t.getMessage(), t);
        }
        else {
            return new MomException(MomConstants.ERR_UNEXPECTED, t.getMessage(), t);
        }
    }

    public static Message toRmqMessage(MomMessage message)
    {
        Message rmqMsg = new Message();
        rmqMsg.setTopic(message.getTopic());
        rmqMsg.setKeys(message.getKey());
        rmqMsg.setTags(message.getTag());
        rmqMsg.setDelayTimeLevel(toRmqDelayLevel(message.getDelayLevel()));
        Iterator<String> names = message.getNames();
        while (names.hasNext()) {
            String name = names.next();
            rmqMsg.putUserProperty(MomConstants.USER_PROPERTY_PREFIX + name, message.getString(name));
        }
        rmqMsg.setBody(SerializationUtils.serialize((Serializable) message.getMsg()));
        return rmqMsg;
    }

    public static MomMessage toMomMessage(MessageExt message)
    {
        MomMessage momMessage = new MomMessage();
        momMessage.setId(message.getMsgId());
        momMessage.setTopic(message.getTopic());
        momMessage.setKey(message.getKeys());
        momMessage.setTag(message.getTags());
        momMessage.setDelayLevel(toMomDelayLevel(message.getDelayTimeLevel()));
        momMessage.setMsg(SerializationUtils.deserialize(message.getBody()));
        Set<String> keys = message.getProperties().keySet();
        for (String key : keys) {
            if (StringUtils.startsWith(key, MomConstants.USER_PROPERTY_PREFIX)) {
                momMessage.setObject(key.substring(MomConstants.USER_PROPERTY_PREFIX.length()),
                        message.getUserProperty(key));
            }
        }
        return momMessage;
    }

    public static MomSendResult toMomResult(SendResult result, MomMessage message)
    {
        MomSendResult mResult = new MomSendResult();
        mResult.setMessage(message);
        mResult.setMsgId(result.getMsgId());
        mResult.setQueueId(String.valueOf(result.getMessageQueue().getQueueId()));
        if (result.getSendStatus() == SendStatus.SEND_OK) {
            mResult.setStatus(MomSendResult.SendStatus.OK);
        }
        else {
            mResult.setStatus(MomSendResult.SendStatus.PARTIAL_OK);
        }
        return mResult;
    }
}
